<?php

error_reporting(E_ALL);
ini_set('display_errors', 'on');

session_start();

include 'config.php';
include 'ZookeeperClient.php';


// 读取本地连接测试结果
if (cache('hosts_utime') && cache('hosts_status') && cache('hosts_utime') > time() - 600) {
    $Hosts_stat = cache('hosts_status');
} else {
    $zookeeper = \timandes\ZookeeperClient::getInstence();
    foreach ($Hosts as $key => $v) {
        try {
            $zookeeper->connect($v['host']);
            $children = $zookeeper->getChildren("/");
            $Hosts_stat[$key]['status'] = true;
        } catch (\ZookeeperConnectionException $zce) {
            $Hosts_stat[$key]['status'] = false;
        }
    }
    cache('hosts_status', $Hosts_stat);
    cache('hosts_utime', time());
}
$Hosts = array_merge_recursive($Hosts, $Hosts_stat);



// 未设置 ver 初始化
$ver = "local";
if (isset($_REQUEST['ver']) && isset($Hosts[$_REQUEST['ver']])) {
    $ver = $_REQUEST['ver'];
    setcookie('ver', $ver);
} elseif(isset($_COOKIE['ver'])){
    $ver = $_COOKIE['ver'];
}
p($ver);
$globalZookeeperHosts = $Hosts[$ver]['host'];


// 设置连接对象为 本次选择对象
try {
    $zookeeper = \timandes\ZookeeperClient::getInstence();
    $zookeeper->connect($globalZookeeperHosts);
    $zookeeper->get("/");
} catch (\ZookeeperClientCoreException $zcce) {
    if ($zcce->getCode() == \ZookeeperClient::ERR_NOAUTH) {
        header("Location: add_auth.php?path=" . urlencode($path));
        exit(1);
    }
    $children = array();
    $getChildrenException = $zcce;
} catch (\ZookeeperConnectionException $zce) {
    echo date("Y-m-d H:i:s") . " [error] 服务器连接异常，请通知管理员！" . " 服务器 [" . $_COOKIE['ver'] . "]";
    exit();
}


/**
 * 缓存数据
 * @param string $key key
 * @param str|array $val val
 * @param string $type 类型默认file
 * @param string $pre 前缀默认空
 * @return 数据
 */
function cache($key, $val=false, $type="file", $pre="") {
    if ($type == "file") {
        $pre = __DIR__."/cache/".$pre."/";
        if (!is_dir($pre)) {
            mkdir($pre, "0777", true);
        }
    }
    $file_name = $pre . $key . ".cache";
    if ($val == "") {
        if (is_file($file_name)) {
            $val = json_decode(file_get_contents($file_name), true);
        }
    } else {
        $val = file_put_contents($file_name, json_encode($val));
    }
    return $val;
}
/**
 * 打印数据
 */
function p() {
    $args = func_get_args();
    foreach ($args as $arg) {
        echo "<pre  style='background-color: #aaa;font-weight: 600;'>".var_export($arg, true)."</pre>";
    }
}

/**
 * 递归地查找最底层数据
 */
function fetch_recursive($path){
    $zookeeper = \timandes\ZookeeperClient::getInstence();
    $children = $zookeeper->getChildren($path);
    if (empty($children)) {
        return $zookeeper->get($path);
    } else {
        foreach ($children as $p) {
            $data[$p] = fetch_recursive($path . "/" . $p);
        }
    }
    return $data;
}
/**
 * 获取 es 数据
 * @param unknown_type $index
 * @param unknown_type $host
 */
function fetch_es_size($index="", $host="10.204.13.112:9200") {
    $url = "http://10.204.13.111:9200/_cat/indices?format=json&pretty";
    $data = json_decode(file_get_contents($url), true);
    foreach ($data as $key => & $item) {
        $index = explode("-", $item['index'], 2);
        if(isset($index[1]) && strtotime($index[1]) > 10) {
            $row = array();
            foreach ($item as $k => $v) {
                $k = str_replace(".","_",$k);
                $row['es_'.$k] = $v;
            }
            $row['topic'] = $index[0];
            $row['cdate'] = $index[1];
            $data[$key] = $row;
        } else{
            unset($data[$key]);
        }
    }
    return $data;
}

/**
 * 保存统计数据
 * @param array $data 统计数据list，每个item必须要有 topic cdate
 * @param string $index 存储的es索引
 * @param string $host host:port
 */
function save_es($data, $index="stat_kafka", $type="stat",$host="10.204.13.112:9200") {
    $es_bulk = $host.'/_bulk';
    $body = '';
    foreach ($data as $item) {
        $body .= '{"create": { "_index": "'.$index.'", "_type": "'.$type.'","_id" : "'.$item['topic'] . $item['cdate'].'"}}' . "\n";
        $body .= json_encode($item) . "\n";
    }
    p($body);
    return fly_curl($es_bulk, $body);
}

function fly_curl($url, $postFields = null) {


    $ch = curl_init();
    curl_setopt($ch, CURLOPT_URL, $url);
    curl_setopt($ch, CURLOPT_USERAGENT, 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:45.0) Gecko/20100101 Firefox/45.0');
    curl_setopt($ch, CURLOPT_HEADER, 0);
    curl_setopt($ch, CURLOPT_FAILONERROR, 0);
    curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
    curl_setopt($ch, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);
    //curl_setopt($ch,CURLOPT_HTTPHEADER,array("Expect:"));
    if (!empty($postFields)) {
         
        $postBodyString = "";
        $postMultipart = false;
         
        if (is_array($postFields) && 0 < count($postFields)) {
            foreach ($postFields as $k => $v)
            {
                if("@" != substr($v, 0, 1))//判断是不是文件上传
                {
                    $postBodyString .= "$k=" . urlencode($v) . "&";
                }
                else//文件上传用multipart/form-data，否则用www-form-urlencoded
                {
                    $postMultipart = true;
                }
            }
            unset($k, $v);
        } else {
            $postBodyString = $postFields . "&";
        }
        curl_setopt($ch, CURLOPT_POST, 1);
        if ($postMultipart)
        {
            curl_setopt($ch, CURLOPT_POSTFIELDS, $postFields);
        }
        else
        {
            //var_dump($postBodyString);
            curl_setopt($ch, CURLOPT_POSTFIELDS, substr($postBodyString,0,-1));
        }
    }
    $reponse = curl_exec($ch);
    // 	echo "<pre>";
    // 	var_dump(curl_getinfo($ch));
    //return curl_getinfo($ch);
    if (curl_errno($ch))
    {
        throw new Exception(curl_error($ch),0);
    }
    else
    {
        $httpStatusCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        if (200 !== $httpStatusCode)
        {
            throw new Exception($reponse,$httpStatusCode);
        }
    }
    curl_close($ch);
    return $reponse;
}